package com.plusmoney.approval.customer;

import com.plusmoney.mq.MQCustomer;
import com.plusmoney.util.Utils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class CustomerUser {
    private static final Logger logger = LoggerFactory.getLogger(CustomerUser.class);

    private static final String MQ_GROUP_NAME = "MP_USER_BROADCAST";
    private static final String MQ_TOPIC = "UserBroadcast";

    private static MQCustomer customer;

    static void start() {
        customer = new MQCustomer();
        customer.group(MQ_GROUP_NAME)
                .topic(MQ_TOPIC)
                .bind();
        try {
            customer.subscribe()
                    .setMessageModel(MessageModel.BROADCASTING)
                    .setConsumeMessageBatchMaxSize(10)
                    .setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET)
                    .listener(new UserListener());

            customer.start();
	
	        logger.info("------CustomerUser start------");
        } catch (MQClientException e) {
	        logger.info("------CustomerUser start fail------");
        logger.error(e.getErrorMessage(),e);
            }
    }

    static void shutdown() {
        logger.info("------CustomerUser shutdown------");
        if (Utils.isNotNull(customer)) {
            customer.shutdown();
        }
    }
}
